#include "batch_job.h"
//# define DEBUG_LEVEL 19
//# define DEBUG_LEVEL 19
#include "debug_func.h"
#include <boost/lexical_cast.hpp>
#include <stdexcept>
#include <cassert>

using std::string;
using std::ostream;
using std::cerr;
using boost::lexical_cast;
//________________________________________________________________________________________

//		get_gene_pairs

//____________________________	____________________________________________________________
bool t_batch_num::get_gene_pairs(	t_gi_pairs& 	gi_pairs,
									ostream&		err_strm)
{
	debug_func_cerr(5);
	string sql_cmd = 	"SELECT "
							"gene_id1, gene_id2 "
						"FROM "
							"batch_gene_pairs "
						"WHERE "
							"batch = " + batch_num_txt + " AND "
							"category = '" + job_category + "'";
	// create binary cursor
	if (!pg_db.Declare(sql_cmd, false))
		return pg_print_sql_err("create a cursor to "
								"retrieve gene pairs for batch =[" + batch_num_txt + "] and "
								"category = ['" + job_category + "']",
								pg_db,
								sql_cmd, err_strm);
	
	{
		unsigned batch_size = 500;
		// get data in batches
		while (pg_db.Fetch(batch_size))
		{
			int curr_batch_size = pg_db.Tuples();
			if (!curr_batch_size)
				break;
			for (int i = 0; i < curr_batch_size; ++i)
			{
				assert(pg_db.getvalue(i, 0));
				assert(pg_db.getvalue(i, 1));
				gi_pairs.push_back(t_gi_pair(pg_db.getvalue(i, 0), pg_db.getvalue(i, 1)));
			}
	
		}
	
	}
	
	// close cursor
	if (!pg_db.Close())
		return pg_print_sql_err("close the  cursor to "
								"retrieve gene pairs for batch =[" + batch_num_txt + "] and "
								"category = ['" + job_category + "']",
								pg_db, "CLOSE CURSOR", err_strm);
	return true;
}



t_batch_num::t_batch_num(t_pg_cursor& pgdb)
	:pg_db(pgdb), committed(true)
{
	debug_func_cerr(5);
}

//________________________________________________________________________________________

//	free_hanging_jobs

//		first see if there are any hanging jobs
//		lock table and remove hanging jobs

//		N.B. Lock Table creates a transaction if one isn't there already

//________________________________________________________________________________________
void t_batch_num::free_hanging_jobs()
{
	debug_func_cerr(5);
	string cmd_SQL =
					"SELECT batch "
						"FROM batch_jobs "
					"WHERE "
						"checked_out = true AND "
						"checked_in = false AND "
						"backend_pid NOT in ("
						"SELECT "
							"pg_stat_get_backend_pid("
								"pg_stat_get_backend_idset())"
						")";
	
	if (!pg_db.ExecTuplesOk(cmd_SQL))
	{
		pg_print_sql_err("free hanging batch jobs in postgreSQL",
							pg_db, cmd_SQL, cerr);
		throw std::runtime_error(string("Postgres"));
	}
	if (pg_db.Tuples() ==0)
		return;
	{
		//t_lock_table lock(pg_db, "lock_transaction");
		t_lock_table lock(pg_db, "batch_jobs");
		string cmd_SQL = "UPDATE batch_jobs "
					"SET "
						"checked_in = false, "
						"checked_out = false, "
						"backend_pid = NULL, "
						"checked_out_when = NULL "
					"WHERE "
						"checked_out = true AND "
						"checked_in = false AND "
						"backend_pid NOT in ("
						"SELECT "
							"pg_stat_get_backend_pid("
								"pg_stat_get_backend_idset())"
						")";
		if (!pg_db.ExecCommandOk(cmd_SQL))
		{
			pg_print_sql_err("remove hanging jobs", pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
	}
}


		
//________________________________________________________________________________________

//	get_next

//		Get next batch number. If fail, free hanging jobs and try again

//________________________________________________________________________________________
bool t_batch_num::get_next(const std::string& job_category_)
{
	debug_func_cerr(5);
	if (!try_get_next(job_category_))
	{
		free_hanging_jobs();
		return try_get_next(job_category_);
	}
	return true;
}

//________________________________________________________________________________________

//	get_next

//		Get next batch number. If fail, free hanging jobs and try again

//________________________________________________________________________________________
bool t_batch_num::get(const std::string& job_category_, unsigned batch_num)
{
	debug_func_cerr(5);
	if (!try_get(job_category_, batch_num))
	{
		free_hanging_jobs();
		return try_get(job_category_, batch_num);
	}
	return true;
}

//________________________________________________________________________________________

//	try_get_next

//		Get next batch number
//		N.B. This takes place within a transaction
//		Dies if another job is opening with my process ID


//________________________________________________________________________________________
bool t_batch_num::try_get_next(const std::string& job_category_)
{
	debug_func_cerr(5);
	// comment this out if the job_category changes mid-stream
	assert (job_category.length() == 0 || job_category_ == job_category);
	
	
	
	t_transaction transaction(pg_db);
	{
		string cmd_SQL =
						"UPDATE batch_jobs "
							"SET "
								"checked_in = false, "
								"checked_out = true, "
								"backend_pid = pg_backend_pid(), "
								"checked_out_when = cast(timeofday() as timestamp) "
							"WHERE "
							"batch_jobs.category = '" + job_category_ + "' AND"
							"batch = ("
										"SELECT batch "
											"FROM batch_jobs "
										"WHERE "
											"checked_out = false AND "
											"category = '" + job_category_ + "' "
										"LIMIT 1)";
				
		if (!pg_db.ExecCommandOk(cmd_SQL))
		{
			pg_print_sql_err("select next batch job from postgreSQL", pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
	}
	
	
	{
		string cmd_SQL =			"SELECT batch "
										"FROM batch_jobs "
									"WHERE "
										"backend_pid = pg_backend_pid() AND "
										"category = '" + job_category_ + "'";


		if (!pg_db.ExecTuplesOk(cmd_SQL))
		{
			pg_print_sql_err("retrieve the next batch todo job from postgreSQL",
								pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
		if (pg_db.Tuples() > 1)
		{
			pg_print_sql_err("retrieve the next batch todo job from postgreSQL. "
								"Another process seems to be running with the same "
								"postgres backend ID.",
								pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
		else if (pg_db.Tuples() ==0)
		{
			pg_print_sql_err("No batch recovered.\nFinished?\n",
								pg_db, cmd_SQL, cerr);
			return false;
		}
		
		assert(pg_db.getvalue(0, 0));
		batch_num_txt = pg_db.GetValue(0, 0);
	}

	
	committed = false;
	job_category = job_category_;
	
	return true;
}

bool t_batch_num::try_get(const std::string& job_category_, unsigned batch_num)
{
	debug_func_cerr(5);
	// comment this out if the job_category changes mid-stream
	assert (job_category.length() == 0 || job_category_ == job_category);
	
	string try_batch_num_txt = lexical_cast<string>(batch_num);
	
	
	t_transaction transaction(pg_db);
	{
		string cmd_SQL =
						"UPDATE batch_jobs "
							"SET "
								"checked_in = false, "
								"checked_out = true, "
								"backend_pid = pg_backend_pid(), "
								"checked_out_when = cast(timeofday() as timestamp) "
							"WHERE "
							"batch_jobs.category = '" + job_category_ + "' AND"
							"batch = " + try_batch_num_txt;
		if (!pg_db.ExecCommandOk(cmd_SQL))
		{
			pg_print_sql_err("select next batch job from postgreSQL", pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
	}
	
	
	{
		string cmd_SQL =			"SELECT batch "
										"FROM batch_jobs "
									"WHERE "
										"backend_pid = pg_backend_pid() AND "
										"category = '" + job_category_ + "'";


		if (!pg_db.ExecTuplesOk(cmd_SQL))
		{
			pg_print_sql_err("retrieve the next batch todo job from postgreSQL",
								pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
		if (pg_db.Tuples() > 1)
		{
			pg_print_sql_err("retrieve the next batch todo job from postgreSQL. "
								"Another process seems to be running with the same "
								"postgres backend ID.",
								pg_db, cmd_SQL, cerr);
			throw std::runtime_error(string("Postgres"));
		}
		else if (pg_db.Tuples() ==0)
		{
			pg_print_sql_err("No batch recovered.\nFinished?\n",
								pg_db, cmd_SQL, cerr);
			return false;
		}
		
		assert(pg_db.getvalue(0, 0));
		batch_num_txt = pg_db.GetValue(0, 0);
	}

	
	committed = false;
	job_category = job_category_;
	
	return true;
}


	

void t_batch_num::commit()
{
	debug_func_cerr(5);
	if (!batch_num_txt.length() || committed)
		return;
	string SQL_command1 = "UPDATE batch_jobs "
							"SET checked_in = TRUE, "
							"checked_in_when = cast(timeofday() as timestamp) "
							"where batch = " + batch_num_txt + " AND "
								"category = '" + job_category + "'";
	if (!pg_db.ExecCommandOk(SQL_command1))
	{
		pg_print_sql_err("check in a batch from the todo list in postgreSQL",
							pg_db, SQL_command1, cerr);
		throw std::runtime_error(string("Postgres"));

	}
	committed = true;
}

const string& t_batch_num::text() const
{
	if (batch_num_txt.length() > 0 && batch_num_txt.length() < 100)
		return batch_num_txt;
	throw std::runtime_error(string("text"));
}
	
	
string get_backend_pid(t_pg_cursor& pg_db)
{
	string select_SQL = "SELECT pg_backend_pid()";
	if (!pg_db.ExecTuplesOk(select_SQL) || pg_db.Tuples() == 0)
	{
		pg_print_sql_err("retrieve the backend pid from postgreSQL",
							pg_db, select_SQL, cerr);
		throw std::runtime_error(string("Postgres"));
	}
	
	assert(pg_db.getvalue(0, 0));
	return (pg_db.GetValue(0, 0));
}



